package com.yuqih.common.websocket.impl;

import java.io.IOException;
import java.util.Collections;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.ExecutorService;

import javax.websocket.Session;

import org.apache.commons.lang3.StringUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.yuqih.common.json.JacksonObjectMapper;
import com.yuqih.common.websocket.IWebSocketIniter;
import com.yuqih.common.websocket.IWebSocketManager;
import com.yuqih.common.websocket.IWebSocketMessageHandler;
import com.yuqih.common.websocket.SessionContainer;
import com.yuqih.common.websocket.WebSocketResponse;

public class WebSocketManager implements IWebSocketManager {
	
	private Log logger = LogFactory.getLog("WebSocket");

	// 这里还应该加一个最后往该session发送消息的时间，如果成功，则更新时间
	// 心跳检测，连续3次不成功，关闭session,从集合中移除该session
	private final Set<SessionContainer> containerSet = new HashSet<SessionContainer>();

	private ExecutorService executorService;
	
	@Override
	public void sendMessage(Session session, WebSocketResponse message) throws IOException {
		if ( message == null) {
			throw new IllegalArgumentException("不能发送空数据");
		}
				
		String text = JacksonObjectMapper.Default_Object_Mapper.writeValueAsString(message);
		
		if (StringUtils.isBlank(text)) {
			throw new IllegalArgumentException("不能发送空数据");
		}
		session.getBasicRemote().sendText(text);
	}
	
	@Override
	public SessionContainer initialize(Session session, IWebSocketMessageHandler handler) {
		return initialize(session,  handler, null);
	}

	@Override
	public SessionContainer initialize(Session session, 
			IWebSocketMessageHandler handler, IWebSocketIniter initer) {
		synchronized (containerSet) {
			for (SessionContainer container : containerSet) {
				// 每开一个websocket，session都是新创建的
				if (container.getSession() == session) {
					throw new IllegalArgumentException("已添加该连接");
				}
			}

			SessionContainer container = new SessionContainer(session, handler, initer);
			containerSet.add(container);

			if (container.getIniter() != null) {
				executorService.execute(new Runnable() {

					@Override
					public void run() {
						try {
							container.getIniter().initialize(container);
						} catch (Exception ex) {
							container.getIniter().handleException(ex);
						}
					}
				});
			}

			executorService.execute(new Runnable() {

				@Override
				public void run() {
					while (container.isRunning()) {
						if (!inited(container)) {
							try {
								Thread.sleep(1000);
							} catch (InterruptedException e) {
								return;
							}
						}

						Object message;
						try {
							message = container.getQueue().take();
						} catch (InterruptedException e) {
							return;
						}

						try {
							container.getHandler().handle(message, container);
						} catch (Exception ignore) {
							logger.error("websocket处理错误", ignore);
						}
					}

				}
			});
			return container;
		}
	}

	private boolean inited(SessionContainer container) {
		if (container.getIniter() == null) {
			return true;
		}

		return container.getIniter().inited();
	}

	public SessionContainer remove(Session session) {
		synchronized (containerSet) {
			Iterator<SessionContainer> iterator = containerSet.iterator();
			while (iterator.hasNext()) {
				SessionContainer container = iterator.next();
				if (container.getSession() == session) {
					iterator.remove();
					container.stop();
					return container;
				}
			}
			return null;
		}

	}

	public Set<SessionContainer> getContainerSet() {		
		return Collections.unmodifiableSet(containerSet);
	}

	/**
	 * @param executorService
	 *            the executorService to set
	 */
	public void setExecutorService(ExecutorService executorService) {
		this.executorService = executorService;
	}

	

}
